package movee.cluster;

import movee.config.EtcdProperties;
import movee.utils.CloseUtils;
import movee.utils.DelayUtils;
import movee.utils.network.LocalHostUtils;
import movee.utils.StringSplitter;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KV;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import jakarta.annotation.PostConstruct;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * 集群管理相关服务
 *
 * @author movee
 */
@Service
@Slf4j
public class ClusterService implements ApplicationRunner, AutoCloseable {

    private final Client etcdClient;
    private final String hostname;
    private final String memberId;
    private final KV kv;
    private final Lease lease;
    private long leaseId = 0L;

    private final AtomicReference<String> leader = new AtomicReference<>("");
    private final String backupNodesDir;

    private final ThreadPoolTaskExecutor taskExecutor;

    private ClusterNodeState clusterState = ClusterNodeState.INIT;
    private final ClusterEventListenerRegistry<ClusterNodeStateListener> listeners;
    private final Map<String, ExecutorService> listenerExecutors;
    private final Object lock = new Object();

    @Autowired
    public ClusterService(EtcdProperties etcdProperties,
                          @Value("${server.port}") int port,
                          @Qualifier(value = "clusterElectionTaskExecutor") ThreadPoolTaskExecutor taskExecutor) {

        // 本节点标识
        this.hostname = LocalHostUtils.getLocalHostname();
        InetAddress localAddress = LocalHostUtils.getLocalAddress();
        this.memberId = localAddress.getHostAddress() + ":" + port;
        this.taskExecutor = taskExecutor;

        this.backupNodesDir = etcdProperties.getBackupClusterNodesDir();

        log.info("current node hostname: {}, memberId: {}", hostname, memberId);

        // 创建etcd客户端实例
        String endpoints = etcdProperties.getEndpoints();
        ByteSequence user = ByteSequence.from(etcdProperties.getUsername(), StandardCharsets.UTF_8);
        ByteSequence passwd = ByteSequence.from(etcdProperties.getPassword(), StandardCharsets.UTF_8);
        this.etcdClient = Client.builder().endpoints(StringSplitter.splitToArrayAndTrim(endpoints, ","))
                .user(user).password(passwd).build();
        this.kv = etcdClient.getKVClient();
        this.lease = etcdClient.getLeaseClient();

        listeners = new SimpleClusterEventListenerRegistry<>();
        listenerExecutors = new HashMap<>();
    }

    @PostConstruct
    public void init() throws Exception {
        // 创建租约，并持续续约
        leaseId = lease.grant(10).get().getID();
        lease.keepAlive(leaseId, new ClusterNodeKeepAliveObserver());
    }

    public void register() {

        try {
            // 注册一个节点，并绑定租约
            PutOption option = PutOption.newBuilder().withLeaseId(leaseId).build();
            ByteSequence key = ByteSequence.from(backupNodesDir + hostname, StandardCharsets.UTF_8);
            ByteSequence value = ByteSequence.from(memberId, StandardCharsets.UTF_8);
            kv.put(key, value, option).get();
        } catch (InterruptedException | ExecutionException e) {
            log.info("register failed. exception info: {}", ExceptionUtils.getStackTrace(e));
        }
    }

    public void unregister() {
        ByteSequence key = ByteSequence.from(backupNodesDir + hostname, StandardCharsets.UTF_8);
        kv.delete(key);
    }

    public String getLeader() {
        return leader.get();
    }

    public boolean selfIsLeader() {
        return Objects.equals(leader.get(), memberId);
    }

    public String getMemberId() {
        return memberId;
    }

    public List<String> getAliveMembers() {
        try {
            ByteSequence key = ByteSequence.from(backupNodesDir, StandardCharsets.UTF_8);
            GetOption getOption = GetOption.newBuilder().isPrefix(true).build();
            return kv.get(key, getOption).get().getKvs().stream()
                    .map(KeyValue::getValue)
                    .map(v -> v.toString(StandardCharsets.UTF_8))
                    .collect(Collectors.toList());
        } catch (InterruptedException | ExecutionException e) {
            log.error("getAliveMembers failed. exception info: {}", ExceptionUtils.getStackTrace(e));
        }
        return new ArrayList<>();
    }

    public ClusterNodeState getClusterState() {
        return clusterState;
    }

    public ClusterInfo getClusterInfo() {
        return new ClusterInfo(clusterState, leader.get(), memberId, getAliveMembers());
    }

    public void elect() {

        try {
            while (true) {

                try {
                    ByteSequence key = ByteSequence.from(backupNodesDir, StandardCharsets.UTF_8);
                    GetOption getOption = GetOption.newBuilder().isPrefix(true).build();
                    long minRevision = Long.MAX_VALUE;
                    KeyValue electedLeader = null;
                    for (KeyValue keyValue : kv.get(key, getOption).get().getKvs()) {
                        if (keyValue.getCreateRevision() < minRevision) {
                            minRevision = keyValue.getCreateRevision();
                            electedLeader = keyValue;
                        }
                    }

                    String currLeader = "";
                    if (electedLeader != null) {
                        currLeader = electedLeader.getValue().toString(StandardCharsets.UTF_8);
                        if (!Objects.equals(currLeader, leader.get())) {
                            log.info("leader changed. new leader info: key: {}, value: {}, createRevision: {}",
                                    electedLeader.getKey().toString(StandardCharsets.UTF_8),
                                    currLeader,
                                    electedLeader.getCreateRevision());
                        }

                        if (!StringUtils.isEmpty(currLeader)) {
                            leader.set(currLeader);
                        }
                    }

                    synchronized (lock) {
                        ClusterNodeState preState = clusterState;
                        if (Objects.equals(currLeader, memberId)) {
                            clusterState = ClusterNodeState.MASTER;
                        } else {
                            clusterState = ClusterNodeState.SLAVE;
                        }
                        notifyIfChanged(preState, clusterState);
                    }

                    DelayUtils.delayBySleep(10000L);
                } catch (ExecutionException ee) {
                    log.warn("catch election exceptio: {}", ExceptionUtils.getStackTrace(ee));
                }
            }
        } catch (Exception e) {
            log.warn("exit elect. exception: {}", ExceptionUtils.getStackTrace(e));
        }
    }

    public void registerClusterNodeStateListener(final String name, final ClusterNodeStateListener listener) {
        synchronized (lock) {
            log.info("register ClusterNodeStateListener, name: {}", name);
            listeners.registerClusterEventListener(name, listener);
            listenerExecutors.put(name, Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
                    .setNameFormat("thread-cluster-event-listener-" + listener.getName() + "-%d")
                    .build()));

            // 通知当前节点状态
            listenerExecutors.get(name)
                    .submit(() -> listener.onClusterNodeStateChange(ClusterNodeState.INIT, clusterState));

        }
    }

    public void removeClusterNodeStateListener(final String name) {
        synchronized (lock) {
            if (!StringUtils.isEmpty(name)) {
                log.info("delete ClusterNodeStateListener, name: {}", name);
                listeners.removeClusterEventListener(name);
                listenerExecutors.remove(name);
            }
        }
    }

    private void notify(final ClusterNodeState previous, final ClusterNodeState current) {

        for (ClusterNodeStateListener listener : listeners) {
            ExecutorService executorService = listenerExecutors.get(listener.getName());
            if (executorService != null) {
                log.info("notify ClusterNodeState changed. name: {}, previous: {}, current: {}",
                        listener.getName(), previous, current);
                executorService.submit(() -> listener.onClusterNodeStateChange(previous, current));
            }

        }
    }

    private void notifyIfChanged(final ClusterNodeState previous, final ClusterNodeState current) {
        if (!Objects.equals(previous, current)) {
            notify(previous, current);
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 注册
        register();

        // 选主
        taskExecutor.execute(this::elect);
    }

    @Override
    public void close() throws Exception {

        // 关闭选主
        taskExecutor.shutdown();

        CloseUtils.close(kv);

        if (leaseId != 0 && Objects.nonNull(lease)) {
            lease.revoke(leaseId);
        }

        CloseUtils.close(lease);
        CloseUtils.close(etcdClient);

    }

    public static class ClusterNodeKeepAliveObserver implements StreamObserver<LeaseKeepAliveResponse> {

        @Override
        public void onNext(LeaseKeepAliveResponse value) {
            log.trace("cluster node lease remaining ttl: {}, lease id: {}", value.getTTL(), value.getID());
        }

        @Override
        public void onError(Throwable t) {
            log.error("cluster node lease keep alive failed. exception info: {}", ExceptionUtils.getStackTrace(t));
        }

        @Override
        public void onCompleted() {
            log.trace("cluster node lease completed");
        }
    }


}
